package com.googlecode.perfectable.prosperable.channel;

import java.io.IOException;
import java.io.InputStream;
import java.io.Reader;
import java.nio.CharBuffer;

import javax.annotation.Nullable;

import org.apache.catalina.websocket.StreamInbound;
import org.apache.catalina.websocket.WsOutbound;

import com.google.common.base.Throwables;
import com.google.common.io.CharStreams;
import com.googlecode.perfectable.eventable.routing.Sink;
import com.googlecode.perfectable.eventable.routing.Source;
import com.googlecode.perfectable.prosperable.message.ClientMessage;
import com.googlecode.perfectable.prosperable.message.ServerMessage;

public class ChannelStreamInbound extends StreamInbound {
	private final Sink<ClientMessage> target;
	private final Source<ServerMessage> source;
	
	private final MessageSerializer serializer;
	
	public ChannelStreamInbound(Source<ServerMessage> source, Sink<ClientMessage> target, MessageSerializer serializer) {
		this.source = source;
		this.target = target;
		this.serializer = serializer;
	}
	
	@Override
	protected void onOpen(@Nullable WsOutbound outbound) {
		if(outbound == null) {
			throw new IllegalArgumentException();
		}
		this.source.listen(ChannelStreamInbound.this::writeMessage);
	}
	
	@Override
	protected void onTextData(@Nullable Reader r) throws IOException {
		if(r == null) {
			throw new IllegalArgumentException();
		}
		final String messageContents = CharStreams.toString(r);
		ClientMessage message = this.serializer.parseClientMessage(messageContents);
		this.target.receive(message);
	}
	
	protected void writeMessage(ServerMessage message) {
		String messageData = this.serializer.serializeServerMessage(message);
		final CharBuffer buffer = CharBuffer.wrap(messageData);
		final WsOutbound outbound = getWsOutbound();
		try {
			outbound.writeTextMessage(buffer);
		}
		catch(IOException e) {
			throw Throwables.propagate(e);
		}
	}
	
	@Override
	protected void onBinaryData(@Nullable InputStream is) throws IOException {
		throw new UnsupportedOperationException();
	}
	
	@Override
	protected void onClose(int status) {
		this.source.stop(ChannelStreamInbound.this::writeMessage);
	}
}
